# coding=utf-8
"""
author = jamon
"""

import asyncio
from contextlib import suppress
from typing import Callable, Awaitable, Any

import websockets

from aiohttp import web

from obespoir.base.common_define import ConnectionStatus
from obespoir.config import Config
# from obespoir.db.mongo import AvailServerConfig
from obespoir.rpcserver.protocol import RpcProtocol
from obespoir.rpcserver.push_protocol import RpcPushProtocol
from obespoir.rpcserver.connection_manager import RpcConnectionManager
from obespoir.server.frame.frame_server import FrameConfig
from obespoir.websocketserver.protocol import WebSocketProtocol
from obespoir.websocketserver.route import WebsocketController
from obespoir.base.global_object import GlobalObject
from obespoir.share.espoir_log import global_log
from obespoir.share.singleton import Singleton

from obespoir.server.frame.frame_manager import FrameManager

try:
    import uvloop  # noqa

    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
except ImportError:
    pass


# async def send_message(remote_name, message):
#     if remote_name not in RpcConnectionManager().conns.keys():
#         return
#     await RpcConnectionManager().send_message(remote_name, message)


class Server(object, metaclass=Singleton):
    """
    启动server主类
    """

    def __init__(self):
        self.host = None
        self.web_handler = {}
        wp = WebSocketProtocol
        self.msg_handler: Callable[[wp, str], Awaitable[Any]] = None  # noqa

        self.loop = asyncio.get_event_loop()

    def register_web_route(self, url, handler):
        """

        :param url: 相对地址， like "/login"
        :param handler:
        :return:
        """
        self.web_handler[url] = handler

    def register_socket_route(self, ws_handler: WebsocketController):
        """

        :param ws_handler: Type[WebsocketHandler]
        :return:
        """
        self.msg_handler = ws_handler.msg_handler
        GlobalObject().after_open_handler = ws_handler.after_open_handler
        GlobalObject().after_close_handler = ws_handler.after_close_handler

    async def start_web(self, web_port):
        """http协议 web"""
        app = web.Application()
        for k, v in self.web_handler.items():
            app.router.add_route("*", k, v)
        runner = web.AppRunner(app)
        await runner.setup()
        site = web.TCPSite(runner, self.host, web_port)
        global_log.info("start", host=self.host, port=web_port)
        await site.start()

    def set_config(self, config: Config):
        """

        :param config:
        :return:
        """
        f_config = FrameConfig()
        f_config.logic_interval = config.frame_broad_interval
        f_config.frame_num = config.frame_num
        GlobalObject().default_frame_config = f_config
        self.host = config.host
        # web_socket port
        ws_port = config.websocket.port if config.websocket else 0
        # web httpserver port
        web_port = config.http.port if config.http else 0
        # rpcserver port
        rpc_port = config.rpc.port if config.rpc else 0
        # remote_ports list
        remote_ports = config.remote_ports
        # if "mongo_uri" == config.available_way:
        #     # 如果高可用使用的是MongoDB存储配置方式
        #     remote_ports = AvailServerConfig.get_instance(
        # uri=config.mongo_uri)
        GlobalObject().update_remote_ports(remote_ports)
        api_path = config.api_path
        if api_path:
            __import__(api_path)

        GlobalObject().loop = self.loop

        GlobalObject().init_from_config(config)

        if ws_port and self.msg_handler:  # websocket server port start
            GlobalObject().ws_server = self.loop.run_until_complete(
                websockets.serve(self.msg_handler, self.host, ws_port,
                                 create_protocol=WebSocketProtocol))

        if web_port and self.web_handler:  # web httpserver port  start
            GlobalObject().http_server = self.loop.run_until_complete(
                self.start_web(web_port))

        if rpc_port:
            GlobalObject().rpc_server = self.loop.run_until_complete(
                self.loop.create_server(RpcProtocol, self.host, rpc_port))

        if remote_ports:
            for rp in remote_ports:
                host = rp.host
                port = rp.port
                s_type = rp.type
                if host and port:
                    RpcConnectionManager().add_type_node(
                        s_type, host, port)
                    RpcConnectionManager().store_connection(host, port, None)
                    if not RpcConnectionManager().get_connection(host, port):
                        remote_serv = self.loop.create_connection(
                            RpcPushProtocol, host=host, port=port)
                        try:
                            self.loop.run_until_complete(remote_serv)
                        except ConnectionRefusedError:
                            global_log.debug("config all conns", conns=RpcConnectionManager().conns)
                            global_log.info("connecting failed", host=host, port=port)

    async def update_remote_rpc_config(self):
        """
        更新远程rpc连接
        :return:
        """
        r = RpcConnectionManager
        for server_type, addr_info in GlobalObject().remote_ports.items():
            remote_names = [r.gen_node_name(t.host, t.port) for t in addr_info]
            for r in r().conns.keys():
                if r not in remote_names:
                    if r().conns[r]["status"] == ConnectionStatus.ESTABLISHED:
                        r().conns[r]["conn"].transport.close()
                    r().conns.pop(r)
            for t in addr_info:
                name = r.gen_node_name(t.host, t.port)
                if name not in r().conns.keys() or \
                        r().conns[name][
                            "status"] != ConnectionStatus.ESTABLISHED:
                    r().add_type_node(server_type.value, t.host, t.port)
                    r().store_connection(t.host, t.port, None)
                    try:
                        await self.loop.create_connection(
                            RpcPushProtocol, host=t.host, port=t.port)
                        global_log.info("success connect", host=t.host, port=t.port)
                    except ConnectionRefusedError:
                        global_log.error("connect failed", host=t.host, port=t.port)

    def start(self, config: Config):
        self.set_config(config)
        # self.loop.run_until_complete(self.schedule())
        asyncio.run_coroutine_threadsafe(self.schedule(), self.loop)
        self.loop.create_task(FrameManager().run())
        # asyncio.run_coroutine_threadsafe(FrameManager().run(), self.loop)
        self.run()

    async def schedule(self):
        # 定时rpc断线重连
        r = RpcConnectionManager
        while True:
            await asyncio.sleep(3)
            # globalLog.info("start new schedule task~")
            for node_type, name_lst in r().type_dict.items():
                for name in name_lst:
                    name_not_exist = name not in r().conns.keys()
                    status = r().conns[name]["status"]
                    no_established = ConnectionStatus.ESTABLISHED != status

                    conn_is_closing = r().conns[name]["conn"] and \
                                      r().conns[name][
                                          "conn"].transport.is_closing()

                    if name_not_exist or no_established or conn_is_closing:
                        host = r().conns[name]["host"]
                        port = r().conns[name]["port"]
                        try:
                            global_log.debug("try to reconnect", name=name, host=host, port=port)
                            await self.loop.create_connection(
                                RpcPushProtocol, host=host, port=port)
                            global_log.info("success connect", host=host, port=port)
                        except ConnectionRefusedError:
                            global_log.error("schedule try connect failed", name=name, host=host, port=port)

    def run(self):
        try:
            self.loop.run_forever()
        finally:
            pending = asyncio.Task.all_tasks()
            for task in pending:
                task.cancel()
                with suppress(asyncio.CancelledError):
                    self.loop.run_until_complete(task)
            if GlobalObject().http_server:
                GlobalObject().http_server.close()
                self.loop.run_until_complete(
                    GlobalObject().http_server.wait_closed())
            if GlobalObject().rpc_server:
                GlobalObject().rpc_server.close()
                self.loop.run_until_complete(
                    GlobalObject().rpc_server.wait_closed())
            if GlobalObject().ws_server:
                GlobalObject().ws_server.close()
                self.loop.run_until_complete(
                    GlobalObject().ws_server.wait_closed())
            self.loop.close()
